package com.kevinwu.zookeepertask.taskone;

import com.kevinwu.zookeepertask.taskone.client.RpcConnect;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import java.util.List;
import java.util.concurrent.CountDownLatch;


/**
 * 客户端项目启动类
 */
@SpringBootApplication
public class ZdyRpcConsumerApplication {

    public static CuratorFramework client;

    private  static CountDownLatch countDownLatch = new CountDownLatch(1);

    public static void main(String[] args) throws Exception {

        SpringApplication.run(ZdyRpcConsumerApplication.class, args);

        List<String> nodePath = getChlidNode("192.168.33.19:2181","/response");
        //获取zookeeper注册的地址和端口，循环建立连接
        int frist = 0; //放置读取的最小值
        int num = 0;//放置最小值的位置
        String[] ipstr = new String[nodePath.size()];
        String[] portstr = new String[nodePath.size()];
        for (int i = 0; i < nodePath.size(); i++) {
            //获取节点的值
            Stat stat = new Stat();
            int tmp = Integer.valueOf(new String(client.getData().storingStatIn(stat).forPath("/response/"+nodePath.get(i))));
            if(frist ==0){
                frist = tmp;
            }else {
                if(tmp<frist){
                    frist = tmp;
                    num = i;
                }
            }
            ipstr[i] = nodePath.get(i).split(":")[0];
            portstr[i] = nodePath.get(i).split(":")[1];

        }
        //启动线程进行连接
        RpcConnect rpcConnect = new RpcConnect(ipstr[num],Integer.valueOf(portstr[num]));
        rpcConnect.start();
        countDownLatch.await();
    }

    /**
     * 监听zookeeper节点的变化
     * @param client
     * @param path
     * @throws Exception
     */
    protected static void watchChild(CuratorFramework client, String path) throws Exception {
        //创建PathChildrenCache
        //参数：true代表缓存数据到本地
        PathChildrenCache pathChildrenCache = new PathChildrenCache(client,path,true);
        //BUILD_INITIAL_CACHE 代表使用同步的方式进行缓存初始化。
        pathChildrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
        pathChildrenCache.getListenable().addListener((cf, event) -> {
            PathChildrenCacheEvent.Type eventType = event.getType();
            switch (eventType) {
                case CONNECTION_RECONNECTED:
                    pathChildrenCache.rebuild();
                    break;
                case CONNECTION_SUSPENDED:
                    break;
                case CONNECTION_LOST:
                    break;
                case CHILD_ADDED:
                    //新增节点后重新进行启动线程连接服务器
                    String strs[] = new String(event.getData().getData()).split(":");
                    RpcConnect rpcConnect = new RpcConnect(strs[0],Integer.valueOf(strs[1]));
                    rpcConnect.start();
                    break;
                case CHILD_UPDATED:
                    break;
                case CHILD_REMOVED:
                    //节点移除
                    String strs1[] = new String(event.getData().getData()).split(":");
                    break;
                default:
            }
        });
    }

    /**
     * 建立zookeeper连接以及增加节点监听
     * @param zkStr
     */
    public static List<String> getChlidNode(String zkStr,String path) throws Exception {

        RetryPolicy exponentialBackoffRetry = new ExponentialBackoffRetry(1000, 3);
        // 使用fluent编程风格
        client = CuratorFrameworkFactory.builder()
                .connectString(zkStr)
                .sessionTimeoutMs(50000)
                .connectionTimeoutMs(30000)
                .retryPolicy(exponentialBackoffRetry)
                .namespace("app")  // 独立的命名空间 /base
                .build();
        client.start();
        Stat stat = client.checkExists().forPath(path);
        if(stat == null){
            client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path);
        }
        List<String> forPath = client.getChildren().forPath(path);
        //注册节点变化监听
        watchChild(client,path);
        return forPath;
    }

}
